package reactor;

import reactor.core.publisher.ConnectableFlux;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

//演示Flux的生成方式
public class FluxGenerate {
    public static void main(String[] args) throws Exception {
        Flux<Integer> flux1 = Flux.just(1, 2, 3);
        Flux<Integer> flux11 = Flux.range(1, 10000);
        Flux<Integer> flux2 = Flux.fromArray(new Integer[]{1, 2, 3});
        Flux<Integer> flux3 = Flux.fromIterable(List.of(1, 2, 3));
        //封装Java8的Stream
        Flux<Integer> flux4 = Flux.fromStream(Stream.of(1, 2, 3));
        //根据某种策略生成Stream，然后再根据相应的Stream生成Flux
        Flux<Integer> flux5 = Flux.fromStream(() -> Stream.of(1, 2, 3));
        //由异步的生产者产生的Flux(会有生产与消费异步进行，无需等到消费完成，才会生产新的内容)
        Flux<Object> flux6 = Flux.create(fluxSink -> {
            for (int i = 0; i < 1000; i++) {
                fluxSink.next(i);
            }
        });

        //由同步生者产生的Flux
        Flux<Object> flux7 = Flux.generate(() -> 0, (state, synSink) -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            if (state == 3) synSink.complete(); //需要手动中止Flux
            synSink.next(System.currentTimeMillis());
            return ++state; //返回新的状态，通知消费者进行消费
        });

        //生成一个热流：hot stream
        ConnectableFlux<Object> flux8 = Flux.create(fluxSink -> {
            while (true) {
                fluxSink.next(System.currentTimeMillis());
            }
        })
                .sample(Duration.ofSeconds(1))
                .publish();
        flux8.subscribe(System.out::println);
        flux8.connect();
    }
}
